package abacab.exec;

import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import abacab.actor.Actor;
import abacab.actor.ActorContextImpl;
import abacab.message.Envelope;
import abacab.queue.ActorQueue;

/**
 * The class handling the dispatching of message
 */
@SuppressWarnings({ "unchecked", "rawtypes" })
public final class DefaultMessageExecutor implements MessageExecutor {
	/** The logger */
	private static final Logger				logger	= LoggerFactory.getLogger(DefaultMessageExecutor.class);

	/** The maximum number of concurrent executors */
	private final int									maxExecutors;

	/** The running tasks */
	private final Queue<ActorRunner>	runners;

	/**
	 * The maximum number of consecutive messages before giving the token back for an other actor's execution
	 */
	private final int									fairness;

	/** The actors to handle */
	private final ActorQueue					actorQueue;

	/** The actors currently queued for execution (or executing) */
	private final Map<Actor, Boolean>	queuedActors;

	/** The current number of concurrent executors */
	private final AtomicInteger				executors;

	/** Indicates if the executor is stopped */
	private volatile boolean					stopped;

	/**
	 * Create an instance
	 * 
	 * @param maxExecutors
	 *          the maximum number of concurrent executors
	 * @param fairness
	 *          the maximum number of consecutive messages before giving the token back for an other actor's execution
	 * @param actorQueue
	 *          the queue holding the actors
	 */
	protected DefaultMessageExecutor(int maxExecutors, int fairness, ActorQueue actorQueue) {
		this.maxExecutors = maxExecutors;
		this.fairness = fairness;
		this.actorQueue = actorQueue;
		this.queuedActors = new IdentityHashMap<Actor, Boolean>();
		this.executors = new AtomicInteger(0);

		this.runners = new ConcurrentLinkedQueue<ActorRunner>();
		for (int i = 0; i < this.maxExecutors; i++) {
			ActorRunner runner = new ActorRunner();
			this.runners.offer(runner);
			runner.start();
		}
	}

	/**
	 * @return the number of queued actors
	 */
	public int getQueuedActors() {
		return this.actorQueue.size();
	}

	/**
	 * @return the fairness
	 */
	public int getMaxConsecutiveMessages() {
		return this.fairness;
	}

	@Override
	public void stop() {
		this.stopped = true;
		Iterator<ActorRunner> itRunners = this.runners.iterator();
		while (itRunners.hasNext()) {
			itRunners.next().terminate();
			itRunners.remove();
		}
	}

	@Override
	public <M> void queueForExecution(Actor<? super M> actor) throws IllegalStateException {
		if (this.stopped) {
			throw new IllegalStateException("Executor stopped");
		}

		Boolean alreadyQueued;
		synchronized (this.queuedActors) {
			alreadyQueued = this.queuedActors.put(actor, Boolean.TRUE);
		}
		if (alreadyQueued != null) {
			// Actor already being scheduled for execution
			return;
		}

		// Check if we could run immediately rather than putting in the queue
		boolean runImmediately;
		int execs;
		do {
			execs = this.executors.get();
			if (execs >= this.maxExecutors) {
				runImmediately = false;
				break;
			}
			runImmediately = true;
		} while (!this.executors.compareAndSet(execs, execs + 1));

		if (runImmediately) {
			if (actor.isReadyForExecution()) {
				execute(actor, null);
				return;
			}

			// Can not execute yet ... enqueue
			this.actorQueue.offer(actor);

			// Try to use our free spot for execution
			nextExecution(null, null);
		}
		else {
			// Must wait before executing ... enqueue
			this.actorQueue.offer(actor);

			// Check if an execution place was freed in the meantime
			do {
				execs = this.executors.get();
				if (execs >= this.maxExecutors) {
					return;
				}
			} while (!this.executors.compareAndSet(execs, execs + 1));

			// Got a free spot to execute => ask for execution
			nextExecution(null, null);
		}
	}

	/**
	 * Ask for the next actor in the queue to be handled
	 * 
	 * @param runner
	 *          the runner that was handling the previous execution (or NULL if none)
	 * @param finishedActor
	 *          the actor with no messages left
	 */
	public <M> void nextExecution(ActorRunner runner, Actor<M> finishedActor) {
		if (finishedActor != null) {
			boolean isEmpty;
			synchronized (this.queuedActors) {
				// Check if a new message has arrived for the finished actor
				isEmpty = finishedActor.getActorContext().getQueue().isEmpty();
				if (isEmpty) {
					if (this.queuedActors.remove(finishedActor) == null) {
						logger.error("Actor " + finishedActor + " was removed concurrently for " + runner);
					}
				}
			}
			if (!isEmpty) {
				if (finishedActor.isReadyForExecution()) {
					// Handle the new message(s)
					return;
				}
				// Could not take the lock back...
				// Requeue the actor
				this.actorQueue.offer(finishedActor);
			}
		}
		Actor actor = this.actorQueue.poll();
		if (actor != null) {
			execute(actor, runner);
		}
		else {
			workFinished(runner);
		}
	}

	/**
	 * Ask for the next actor in the queue to be handled
	 * 
	 * @param runner
	 *          the runner that was handling the previous execution (or NULL if none)
	 * @param postponedActor
	 *          the actor to reschedule
	 */
	public <M> void postponedExecution(ActorRunner runner, Actor<M> postponedActor) {
		Actor actor = this.actorQueue.poll();
		if (actor == null) {
			if (postponedActor.isReadyForExecution()) {
				// Continue handling of the postponed actor (we took the lock back))
				// nothing to do ;o)
				return;
			}
			// Could not re-take the lock on the postponed actor
			this.actorQueue.offer(postponedActor);

			// See if we have a new comer
			actor = this.actorQueue.poll();
			if (actor == null) {
				workFinished(runner);
				return;
			}
		}
		else {
			// Re-queue the actor
			this.actorQueue.offer(postponedActor);
		}
		execute(actor, runner);
	}

	/**
	 * Handle the point when a runner can execute no actor. It must free the execution spot.
	 * 
	 * @param runner
	 *          the runner
	 */
	private void workFinished(ActorRunner runner) {
		if (this.executors.decrementAndGet() == 0) {
			// If it was the last executor, double check if we can not find a new actor to execute
			Actor actor = this.actorQueue.poll();
			NewComer:
			if (actor != null) {
				// A new comer => try to take the execution spot back
				int execs;
				do {
					execs = this.executors.get();
					if (execs >= this.maxExecutors) {
						// Could not take the execution spot back
						// Put the actor back in the queue
						// The poor actor lost his spot as head of the list
						// Should be extremely rare though ;o) - huge concurrency
						this.actorQueue.offer(actor);
						actor = null;
						break NewComer;
					}
				} while (!this.executors.compareAndSet(execs, execs + 1));

				// Launch the execution
				execute(actor, runner);
				return;
			}
		}

		// No message can be executed for now (stop executing)
		if (runner != null) {
			runner.free();
			if (this.stopped) {
				runner.terminate();
			}
			else {
				this.runners.offer(runner);
			}
		}
	}

	/**
	 * Manages the execution of a message
	 * 
	 * @param actor
	 *          the actor
	 * @param previousRunner
	 *          the runner that was handling the previous execution (or NULL if none)
	 */
	private <M> void execute(Actor<M> actor, ActorRunner previousRunner) {
		ActorContextImpl actorCtx = (ActorContextImpl) actor.getActorContext();
		ActorRunner runner = actorCtx.getRunner();
		if (runner == null) {
			// No runner set
			ActorRunner r;
			if (previousRunner != null) {
				previousRunner.nextActor(actor);
				r = previousRunner;
			}
			else {
				do {
					// The polled runner can be concurrently handled by a runner asking for it
					// Need to check if it can handle this actor
					r = this.runners.poll();
				} while (!r.handle(actor));
			}
			// if (actor.isSticky()) {
			actorCtx.setRunner(r);
			// }
		}
		else if (runner == previousRunner) {
			previousRunner.nextActor(actor);
		}
		else if (runner.handle(actor)) {
			if (previousRunner != null) {
				previousRunner.free();
				this.runners.offer(previousRunner);
			}
			this.runners.remove(runner);
		}
		else {
			// Runner is currently busy
			// Find a runner that will accept the message
			ActorRunner r;
			if (previousRunner != null) {
				previousRunner.nextActor(actor);
				r = previousRunner;
			}
			else {
				do {
					// The polled runner can be concurrently handled by a runner asking for it
					// Need to check if it can handle this actor
					r = this.runners.poll();
				} while (!r.handle(actor));
			}
			actorCtx.setRunner(r);
		}
	}

	//
	//
	//
	//
	//
	//
	//
	//
	//
	//
	// ************************************************************************
	// ************************************************************************
	// THE RUNNER
	// ************************************************************************
	// ************************************************************************

	/**
	 * Handles the execution of an actor
	 */
	public final class ActorRunner extends Thread {
		/** The actor currently running */
		private final AtomicReference<Actor>	currentActor;

		/** Indicates if the executor was stopped */
		@SuppressWarnings("hiding")
		private volatile boolean							stopped;

		/**
		 * Create an instance
		 */
		protected ActorRunner() {
			this.currentActor = new AtomicReference<Actor>(null);
		}

		/**
		 * Stop the executor
		 */
		protected void terminate() {
			this.stopped = true;
			synchronized (this.currentActor) {
				this.currentActor.notify();
			}
		}

		/**
		 * Ask the runner to handle an actor's message queue
		 * 
		 * @param actor
		 *          the actor
		 * @return true if the runner accepts to handle the actor
		 */
		protected <M> boolean handle(Actor actor) {
			if (this.currentActor.compareAndSet(null, actor)) {
				if (Thread.currentThread() != this) {
					// Notify the other thread to wake up
					synchronized (this.currentActor) {
						this.currentActor.notify();
					}
				}
				return true;
			}
			return false;
		}

		/**
		 * Free the runner
		 */
		protected void free() {
			this.currentActor.set(null);
		}

		/**
		 * Ask the runner to handle a new actor. Call this method instead of {@link #free()} and {@link #handle(Actor)} to prevent an actor that barges in.<br />
		 * This method should only be called by the runner's thread.
		 * 
		 * @param actor
		 *          the new actor to execute
		 */
		protected void nextActor(Actor actor) {
			this.currentActor.set(actor);
		}

		@Override
		@SuppressWarnings("synthetic-access")
		public void run() {
			try {
				while (true) {
					Actor actor = this.currentActor.get();
					if (actor != null) {
						Queue<Envelope> queue = actor.getActorContext().getQueue();
						Envelope message;
						int i = 0;
						while ((message = queue.poll()) != null) {
							try {
								actor.onMessage(message);
							}
							catch (Throwable e) {
								logger.error("The handling of a message failed [actor=" + actor + ", message=" + message + "]", e);
							}
							i++;
							if (i >= getMaxConsecutiveMessages()) {
								break;
							}
						}

						// Notify that the execution is finished
						actor.afterExecution();

						// Notify that we are finished (asks for a new actor to execute - potentially in the current runner)
						if (queue.isEmpty()) {
							nextExecution(this, actor);
						}
						else {
							postponedExecution(this, actor);
						}
					}
					else {
						synchronized (this.currentActor) {
							actor = this.currentActor.get();
							if (actor == null) {
								if (this.stopped) {
									// Stop only once all messages have been handled
									return;
								}
								this.currentActor.wait();
							}
						}
					}
				}
			}
			catch (InterruptedException e) {
				// Can safely ignore since the thread is now ending ;o)
			}
		}
	}
}
